package com.ruoyi.flink.wc;

import com.alibaba.fastjson.JSONObject;
import com.ruoyi.flink.wc.entity.Student;
import com.ruoyi.flink.wc.util.RocketMQSource;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.util.Collector;

/**
 * @Author syc
 * @Date Created in 10:24 2022/5/6
 * @Description 从 rocketmq中获取年龄字段, 根据年龄从mysql中获取该年龄的学生,
 *  向该年龄段的孩子,发出生日祝福
 */
public class FlinkMysqlDemo {
    public static void main(String[] args) throws Exception {
       // System.setProperty("rocketmq.client.logLevel", "ERROR");
        // 参数中获取值
        final ParameterTool parameterTool = ParameterTool.fromArgs(args);
        // 获取 rocketmq的地址端口  "192.168.1.125:9876,192.168.1.125:9875"
        String namesrvaddr = parameterTool.get("namesrvaddr", "192.168.1.125:9876");
        // 获取主题(topic)
        String topic = parameterTool.get("topic", "flink-test");
        // 获取分组(group)
        String group = parameterTool.get("flinkGroup", "flink-group");
        // 创建流式执行环境
        StreamExecutionEnvironment env =  StreamExecutionEnvironment.getExecutionEnvironment();
        // 设置并行度(默认就是以)
        env.setParallelism(1);
        // 创建表环境
      //  StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
       // 获取mq中的参数值
        DataStreamSource<String> mqSource = env.addSource(new RocketMQSource(namesrvaddr, topic, group));

        SingleOutputStreamOperator<Integer>
                wordAndOne =  mqSource.flatMap(new FlatMapFunction<String, Integer>() {
            @Override
            public void flatMap(String source, Collector<Integer> collector) throws Exception {
                   try {
                       JSONObject json = JSONObject.parseObject(source);
                       Integer age = json.getInteger("age");
                       if(age != null) {
                          // DataStreamSource<Student> mapQyxxJdbcRead = env.addSource(new FlinkMysqlUtil(age));

                           collector.collect(age);


                          }

                   } catch (Exception ex) {

                   }
            }
        });
       // wordAndOne.print();
        wordAndOne.addSink(new RichSinkFunction<Integer>(){
            @Override
            public void invoke(Integer age, Context context) {
                System.out.println(age+ "..........1111111111");
            }
        });


      //  DataStreamSource<Student> mapQyxxJdbcRead = env.addSource(new FlinkMysqlUtil(12));

      //  mapQyxxJdbcRead.print();
        // 7. 执行
        env.execute();
    }


}
